package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.heima.common.constant.ScheduleConstant;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.enums.TaskStatus;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.*;


@Slf4j
@Service
public class TaskServiceImpl extends ServiceImpl<TaskinfoMapper,Taskinfo>implements TaskService {


    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedissonClient redissonClient;
        /**
     * 添加任务到数据库和Redis缓存的主流程方法
     *
     * 该方法实现任务持久化到MySQL数据库和Redis缓存的双写操作，包含以下核心步骤：
     * 1. 调用addTaskToDb执行数据库持久化（主表+日志表双写）
     * 2. 当数据库操作成功时，根据任务执行时间将任务写入Redis不同数据结构：
     *    - 即时任务（执行时间≤当前时间）存入List队列
     *    - 延迟任务（5分钟内）存入ZSet有序集合
     *
     * @param task 待添加的任务对象，必须包含以下必要信息：
     *             - taskType 任务类型（决定Redis存储key）
     *             - priority 优先级（决定Redis存储key）
     *             - executeTime 执行时间戳（毫秒级）
     * @return 返回生成的任务ID，若数据库操作失败返回0
     * @throws Transactional 注解保证数据库和Redis操作的事务性
     *         当数据库操作失败时会触发事务回滚（通过TransactionAspectSupport实现）
     */
    @Override
    @Transactional
    public long addTask(Task task) {
        // 添加任务到数据库中
        Boolean success = addTaskToDb(task);
        // 添加任务到Redis中
        if (success){
            addTaskToRedis(task);
        }
        return task.getTaskId();
    }

    /**
     * Redis缓存持久化专用方法
     *
     * 根据任务执行时间采用分级存储策略：
     * 1. 即时任务：executeTime ≤ 当前时间 → List结构（立即消费队列）
     * 2. 延迟任务：当前时间 < executeTime ≤ 未来5分钟 → ZSet结构（按时间排序）
     *
     * 存储结构设计说明：
     * - key格式："{taskType}_{priority}"
     * - 未来任务存储空间：ScheduleConstant.FUTURE + key
     * - 即时任务存储空间：ScheduleConstant.TOPIC + key
     *
     * @param task 包含任务ID的任务对象（需确保taskinfoMapper可查询到完整信息）
     * @see #addTask(Task) 主流程方法调用点
     */
    private void addTaskToRedis(Task task){
        Taskinfo taskinfo = taskinfoMapper.selectById(task.getTaskId());
        String key = taskinfo.getTaskType()+"_"+taskinfo.getPriority();

        Instant now = Instant.now();
        Instant plus = now.plus(5, ChronoUnit.MINUTES);
        // 如果任务执行时间小于等于当前时间，存list
        if (taskinfo.getExecuteTime().getTime()<=now.toEpochMilli()){
            stringRedisTemplate.opsForList().leftPush(ScheduleConstant.TOPIC + key, JSON.toJSONString(taskinfo));
        }
        // 如果任务执行时间大于当前时间，小于等于预设时间（未来5分钟）存入zset中
        else if (taskinfo.getExecuteTime().getTime()<=plus.toEpochMilli()){
            stringRedisTemplate.opsForZSet().add(ScheduleConstant.FUTURE + key,JSON.toJSONString(taskinfo),taskinfo.getExecuteTime().getTime());
        }
    }

    /**
     * 数据库持久化核心方法
     *
     * 实现任务信息的双表写入操作：
     * 1. 主任务表(taskinfo)写入：
     *    - 属性拷贝（BeanUtils）
     *    - 时间格式转换（long → Date）
     *    - 自增主键回填
     * 2. 任务日志表(taskinfoLogs)写入：
     *    - 属性拷贝（BeanUtils）
     *    - 初始化版本号（version=1）
     *    - 初始状态设置（INIT）
     *
     * @param task 待持久化的任务对象（包含必要业务数据）
     * @return 操作结果：
     *         - true：双表写入成功
     *         - false：任一表写入失败（触发事务回滚）
     * @exception Exception 捕获所有异常并记录错误日志 通过TransactionAspectSupport实现事务回滚
     * @see TaskStatus 任务状态枚举定义
     */
    private Boolean addTaskToDb(Task task) {
        try{
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoMapper.insert(taskinfo);
            task.setTaskId(taskinfo.getTaskId());

            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo,taskinfoLogs);
            taskinfoLogs.setVersion(1);
            taskinfoLogs.setStatus(TaskStatus.INIT);
            taskinfoLogsMapper.insert(taskinfoLogs);
            return true;
        }catch (Exception e){
            log.error("addTaskToDb fail:{}",e);
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
        }
        return false;
    }





    /**
     * 取消任务
     * @param taskId
     * @return
     */
    @Override
    @Transactional
    public boolean cancelTask(long taskId) {
        //删除任务，更新日志
        Task task = updateDb(taskId,TaskStatus.CANCELLED);
        //删除redis的数据
        if(task != null){
            removeTaskFromRedis(task);
            return true;
        }
        return false;
    }

    /**
     * 删除redis中的任务数据
     * @param task
     */
    private void removeTaskFromRedis(Task task) {

        String key = task.getTaskType()+"_"+task.getPriority();

        if(task.getExecuteTime()<=System.currentTimeMillis()){
            stringRedisTemplate.opsForList().remove(ScheduleConstant.TOPIC+key,0,JSON.toJSONString(task));
        }else {
            stringRedisTemplate.opsForZSet().remove(ScheduleConstant.FUTURE+key,JSON.toJSONString(task));
        }
    }

    /**
     * 删除任务，更新任务日志状态
     * @param taskId
     * @param status
     * @return
     */
    private Task updateDb(long taskId, TaskStatus status) {
        Task task = null;
        try {
            //删除任务
            taskinfoMapper.deleteById(taskId);

            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
            taskinfoLogs.setStatus(status);
            taskinfoLogsMapper.updateById(taskinfoLogs);

            task = new Task();
            BeanUtils.copyProperties(taskinfoLogs,task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        }catch (Exception e){
            log.error("updateDb fail:{}",e);
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
        }

        return task;

    }



    /**
     * 按照类型和优先级拉取任务
     * @return
     */
    @Override
    public Task poll(int type,int priority) {
        Task task = null;
        try {
            String key = type+"_"+priority;
//            String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
            String taskJson = stringRedisTemplate.opsForList().rightPop(ScheduleConstant.TOPIC + key);
            if(StringUtils.isNotBlank(taskJson)){
                task = JSON.parseObject(taskJson, Task.class);
                //更新数据库信息
                updateDb(task.getTaskId(),TaskStatus.EXECUTED);
            }
        }catch (Exception e){
            log.error("poll task exception");
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
        }
        return task;
    }
    public List<String> scan(String pattern){
        return stringRedisTemplate.execute(
                (RedisCallback<List<String>>) connection -> {
                    List<String> matchedKeys = new ArrayList<>();

                    // 构建Redis SCAN操作的配置参数
                    // ScanOptions.builder()链式构建：
                    // 1. match(pattern): 设置键匹配规则
                    // 2. count(count): 指定每次扫描的样本数量
                    // 3. build(): 生成最终的扫描配置对象
                    ScanOptions scanOptions = ScanOptions.scanOptions()
                            .match(pattern)
                            .count(10000)
                            .build();

                    // 初始化游标扫描并处理结果集
                    // try-with-resources自动管理游标资源释放
                    // Cursor<byte[]>: Redis返回的原始字节游标
                    try (Cursor<byte[]> cursor = connection.scan(scanOptions)) {
                        // 遍历游标获取所有匹配键
                        // hasNext(): 检查是否还有剩余元素
                        // next(): 获取下一批量的键数据
                        while (cursor.hasNext()) {
                            byte[] keyBytes = cursor.next();
                            // 将Redis返回的原始字节数组转换为字符串
                            // 使用StandardCharsets.UTF_8确保编码一致性
                            String key = new String(keyBytes, StandardCharsets.UTF_8);
                            matchedKeys.add(key);
                        }
                    } catch (Exception e) {
                        // 包装并抛出扫描异常
                        // 包含原始异常信息和堆栈跟踪
                        // 保证异常诊断信息的完整性
                        throw new RuntimeException("Scan keys failed: " + e.getMessage(), e);
                    }

                    return matchedKeys;
                }
        );
    }


    @Scheduled(cron = "0 */1 * * * ?")
    public void refresh(){
        log.info("定时任务---未来任务定时刷新 开始");
        RLock lock = redissonClient.getLock("FUTURE_TASK_SYNC");
        boolean tryLock = lock.tryLock();
        if (tryLock){
            List<String> futureKeys = scan(ScheduleConstant.FUTURE + "*");
            for (String futureKey:futureKeys){
                //当前未来key的topickey
                String topicKey = futureKey.replace(ScheduleConstant.FUTURE,ScheduleConstant.TOPIC);
                Set<String> tasks = stringRedisTemplate.opsForZSet().rangeByScore(futureKey, 0, Instant.now().toEpochMilli());
                if (tasks!=null&&!tasks.isEmpty()){
                    stringRedisTemplate.executePipelined((RedisCallback<Void>) connection -> {
                        StringRedisConnection stringRedisConnection = (StringRedisConnection) connection;
                        String[] array = tasks.toArray(new String[0]);
                        stringRedisConnection.rPush(topicKey,array);
                        stringRedisConnection.zRem(futureKey,array);
                        return null;
                    });
                    log.info("刷新未来任务：{}，到任务队列：{}",futureKey,topicKey);
                }
            }
        lock.unlock();
        }
        log.info("定时任务---未来任务定时刷新 结束");
    }

    @Scheduled(cron = "0 */5 * * * ?")
    @PostConstruct
    public void reloadData() {
        log.info("定时任务---数据库数据同步到缓存 开始");
        RLock lock = redissonClient.getLock("RELOAD_TASK_SYNC");
        boolean tryLock = lock.tryLock();
        if (tryLock) {
            clearCache();
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, 5);

            //查看小于未来5分钟的所有任务
            List<Taskinfo> allTasks = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime, calendar.getTime()));
            if (allTasks != null && !allTasks.isEmpty()) {
                for (Taskinfo taskinfo : allTasks) {
                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfo, task);
                    task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                    addTaskToRedis(task);
                    log.info("同步数据库数据到Redis：{}，",taskinfo);
                }
            }
            lock.unlock();
        }
        log.info("定时任务---数据库数据同步到缓存 结束");
    }

    private void clearCache(){
        // 删除缓存中未来数据集合和当前消费者队列的所有key
        // future_
        List<String> futurekeys = scan(ScheduleConstant.FUTURE + "*");
        // topic_
        List<String> topickeys = scan(ScheduleConstant.TOPIC + "*");
        stringRedisTemplate.delete(futurekeys);
        stringRedisTemplate.delete(topickeys);
    }
}

